package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.function.process.OdsConnectDataProcessFunction;
import cn.gwm.flink.streaming.ods.ModelHandle;
import cn.gwm.flink.streaming.ods.model.StandardFieldConf;
import cn.gwm.flink.streaming.sink.hbase.HbaseSinkClient;
import cn.gwm.flink.streaming.sink.kafka.KafkaSinkProducer;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.util.StrUtil;
import cn.hutool.db.Db;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ods 接云服务任务
 * @author GW00256253
 */
public class OdsYunTask {
    private static List<String> semaphoreList;
    static {
        semaphoreList = new ArrayList<>();
        try {
            Db.use(DSFactory.get(ConfigLoader.get("mysql.group.db.config")))
                    .find(StandardFieldConf.getEntityWhere(), StandardFieldConf.class).stream()
                    .filter(data -> ModelHandle.BIZ_TYPE_ODS.equals(data.getBizType())
                            &&Integer.parseInt(data.getState())>0
                            &&"2".equals(data.getVersion())
                            &&VehicleType.P03_PHEV.equals(data.getVehicletype()))
                    .sorted(new Comparator<StandardFieldConf>() {
                        @Override
                        public int compare(StandardFieldConf o1, StandardFieldConf o2) {
                            return Integer.parseInt(o1.getId())-Integer.parseInt(o2.getId());
                        }
                    })
                    .forEach(data -> semaphoreList.add(data.getSemaphore()));
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        setEnv(env,"OdsYunTask");
        OutputTag<JSONObject> gpsTag = new OutputTag<JSONObject>("gps_semaphore"){};
        OutputTag<JSONObject> gpsTag1 = new OutputTag<JSONObject>("gps1_semaphore"){};
        OutputTag<JSONObject> hbaseTag1 = new OutputTag<JSONObject>("gps1_hbaseTag_semaphore"){};
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>("gps_hbaseTag_semaphore"){};
        OutputTag<JSONObject> reqTag = new OutputTag<JSONObject>("reqTag_semaphore"){};
        SingleOutputStreamOperator<JSONObject> stream = env.addSource(getKafkaConsumer()).map(JSONUtil::parseObj)
                .filter(item -> BaseFields.COMMAND_STATUS.equals(item.getStr(BaseFields.command))
                        || BaseFields.COMMAND_TBOX.equals(item.getStr(BaseFields.command))
                        || BaseFields.COMMAND_REMOTE_REQUEST.equals(item.getStr(BaseFields.command))
                        || BaseFields.COMMAND_REMOTE_CONTROL.equals(item.getStr(BaseFields.command))
                    ).process(new TranBigDataStatus(VehicleType.P03_PHEV, gpsTag,hbaseTag));
        SingleOutputStreamOperator<JSONObject> canStream = stream.keyBy(entries -> entries.getStr(BaseFields.vin))
                .process(new OdsConnectDataProcessFunction(gpsTag1,hbaseTag1,reqTag)).name("connectData").uid("connectData");
        canStream.addSink(BaseTask.getOrcTxtSink("ods_can_p03","ods/yun/", semaphoreList)).name("toHiveP03Can")
                .uid("toHiveP03Can");;
        canStream.addSink(getProducer()).name("toKafkaP03");
        stream.getSideOutput(gpsTag)
                .addSink(BaseTask.getOrcTxtSink("ods_gps_p03","ods/yun/", BeanSource.SourceEnum.gpsIntranet))
                .name("toHiveP03Gps").uid("toHiveP03Gps");
        stream.getSideOutput(hbaseTag)
                .addSink(BaseTask.getOrcTxtSink("ods_req_p03","ods/yun/", BeanSource.SourceEnum.remoteRequest))
                .name("toHiveP03Request").uid("toHiveP03Request");
//        HbaseSinkClient.hbaseSink(stream.getSideOutput(hbaseTag), ConfigLoader.get("hbase.ods.table.name.gps.p03"),
//                ConfigLoader.get("hbase.table.columnfamily"),"toHbaseP03");
        env.execute("OdsYunTask");
    }

    private static FlinkKafkaProducer getProducer(){
        Properties prodProps = new Properties();
        prodProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ConfigLoader.get("bootstrap.servers"));
        prodProps.setProperty(ProducerConfig.ACKS_CONFIG, "all");
//        //发送频率
        prodProps.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1000");
        //批量发送
        prodProps.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "300000");
        // 5m
        prodProps.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "5242880");
        String suffix = System.getenv().get(DefaultConstant.BOOT_PROFILES_ACTIVE);
        if (suffix == null) {
            suffix = "dev";
        }
        FlinkKafkaProducer  producer;
        if("dev".equals(suffix)){
            producer= KafkaSinkProducer.producerBuilder("ods_can_status_data", prodProps,BaseFields.vin);
        }else{
            producer = KafkaSinkProducer.producerBuilder("ods_phev", prodProps,BaseFields.vin);
        }
        return producer;
    }

    private static FlinkKafkaConsumer<String> getKafkaConsumer(){
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ConfigLoader.get("yun.server"));
        String suffix = System.getenv().get(DefaultConstant.BOOT_PROFILES_ACTIVE);
        if (suffix == null) {
            suffix = "dev";
        }
        if("dev".equals(suffix)) {
            props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-" + ConfigLoader.get("yun.p03.topic")+"-dev");
        } else{
            props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-" + ConfigLoader.get("yun.p03.topic")
                    + suffix + "-p03");
        }
        //如果value合法，则自动提交偏移量
        props.put("enable.auto.commit", "true");
        //设置多久一次更新被消费消息的偏移量
        props.put("auto.commit.interval.ms", "1000");
        //设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息
        props.put("session.timeout.ms", "30000");
        //自动重置offset
        props.put("auto.offset.reset","earliest");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        final FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                ConfigLoader.get("yun.p03.topic"),
                new SimpleStringSchema(),
                props
        );
        kafkaConsumer.setStartFromGroupOffsets();
//        kafkaConsumer.setStartFromTimestamp(1682564400000L);
        return kafkaConsumer;
    }

    private static void setEnv(StreamExecutionEnvironment env,String taskName){
        // 3.1）设置每隔30s一个周期开启checkpoint（checkpoint开启周期，不能太长也不能太短：1-5分钟）
        env.enableCheckpointing(300 * 1000L);
        // 启动类型
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        // 3.2）设置检查点的mode，exactly-once，保证数据的一次性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 3.3）设置两次checkpoint之间的时间间隔，避免两次间隔时间太近导致频繁的checkpoint
        env.getCheckpointConfig().setCheckpointInterval(300*1000L);
        // 3.4）设置checkpoint的超时时间（一般设置一个checkpoint周期的二分之一）
        env.getCheckpointConfig().setCheckpointTimeout(600 * 1000L);
        // 3.5）设置checkpoint的最大尝试次数，同一个时间内有几个checkpoint在运行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        // 3.6）设置job作业取消的时候，是否保留checkpoint的计算结果
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 3.7）设置job作业运行过程中，如果checkpoint失败，是否停止作业
        env.getCheckpointConfig().setFailOnCheckpointingErrors(true);
        // 3.8）设置checkpoint的存储后端，使用rocksdb作为状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        // 设置可容忍的检查点的失败数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        //启用不对齐检查点保存方式
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        env.getCheckpointConfig().setCheckpointStorage(ConfigLoader.get("hdfsUri") + "/flink/checkpoints/" + taskName);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.minutes(15),Time.seconds(10)));
    }

    /**
     *@Author xzh
     *@Description 解析原始数据
     *@Date 2023/1/11 11:05
     *@Param
     *@return
     **/
    static class TranBigDataStatus extends  ProcessFunction<JSONObject, JSONObject> {
        private Logger log = LoggerFactory.getLogger(TranBigDataStatus.class);

        private String vehicleType;

        private OutputTag<JSONObject> gpsTag;

        private OutputTag<JSONObject> hbaseTag;

        private Map<String,String> subCodeSemaphore = new ConcurrentHashMap<>();

        private TranBigDataStatus(String vehicleType,OutputTag<JSONObject> gpsTag,OutputTag<JSONObject> hbaseTag){
            this.vehicleType = vehicleType;
            this.gpsTag = gpsTag;
            this.hbaseTag = hbaseTag;
            try {
                Db.use(DSFactory.get(ConfigLoader.get("mysql.group.db.config"))).find(StandardFieldConf.getEntityWhere(), StandardFieldConf.class).stream()
                        .filter(data -> ModelHandle.BIZ_TYPE_ODS.equals(data.getBizType())&&"1".equals(data.getState())&&this.vehicleType.equals(data.getVehicletype()))
                        .forEach(data -> subCodeSemaphore.put(data.getSubCode(),data.getSemaphore()));
                log.info("读取mysql 表standard_field_conf 成功");
            } catch (SQLException e) {
                log.error("读取mysql 表standard_field_conf 失败");
                throw new RuntimeException(e);
            }
        }

        @Override
        public void processElement(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> collector) throws Exception {
            try {
                List<JSONObject> beanList = object.getBeanList(BaseFields.BIG_DATA_STATUS, JSONObject.class);
                //key为item_time ,value为（字段：值）对象
                Map<String, List<JSONObject>> map = new HashMap<>(64);
                if (beanList != null && beanList.size() > 0) {
                    for (JSONObject bigDataStatus : beanList) {
                        String codeValue = bigDataStatus.getStr(BaseFields.BIG_DATA_CODE);
                        //过滤我们需要的字段
                        if (subCodeSemaphore.containsKey(codeValue)) {
                            List<JSONObject> itemList = bigDataStatus.getBeanList(BaseFields.BIG_DATA_ITEMS, JSONObject.class);
                            if (itemList != null && itemList.size() > 0) {
                                for (JSONObject item : itemList) {
                                    Long aLong = item.getLong(BaseFields.BIG_DATA_ITEMS_TIME);
                                    if (aLong != null && aLong.longValue() > 0) {
                                        //item_time 提取
                                        String key = String.valueOf(aLong.longValue() / 1000 * 1000);
                                        List<JSONObject> objectList = map.get(key);
                                        JSONObject obj = JSONUtil.createObj();
                                        obj.set(BaseFields.BIG_DATA_CODE, subCodeSemaphore.get(codeValue));
                                        obj.set(BaseFields.BIG_DATA_ITEMS_VALUE, item.getStr(BaseFields.BIG_DATA_ITEMS_VALUE));
                                        if (objectList == null) {
                                            objectList = new ArrayList<>();
                                        }
                                        objectList.add(obj);
                                        map.put(key, objectList);
                                    }
                                }
                            }
                        }
                    }
                }
                if (map.size() > 0) {
                    String item_time = object.getStr("acquisitionTime");
                    int listLen = 0;
                    for (Map.Entry<String, List<JSONObject>> set : map.entrySet()) {
                        if (set.getValue().size() >= listLen) {
                            listLen = set.getValue().size();
                            item_time = set.getKey();
                        }
                        //gps要所有数据
                        if (BaseFields.COMMAND_TBOX.equals(object.getStr(BaseFields.command))) {
                            for (JSONObject obj : set.getValue()) {
                                object.set(obj.getStr(BaseFields.BIG_DATA_CODE), obj.getStr(BaseFields.BIG_DATA_ITEMS_VALUE));
                            }
                            object.set(BaseFields.tid, set.getKey());
                            object.remove(BaseFields.BIG_DATA_STATUS);
                            object.set(VehicleType.VEHICLE_TYPE, this.vehicleType);
                            object.set(BaseFields.STORE_TIME, System.currentTimeMillis());
                            collector.collect(object);
                            ctx.output(gpsTag, object);

                        }
                    }
                    //can信号，只要最新的一条数据
                    if (BaseFields.COMMAND_STATUS.equals(object.getStr(BaseFields.command))) {
                        List<JSONObject> objectList = map.get(item_time);
                        for (JSONObject obj : objectList) {
                            object.set(obj.getStr(BaseFields.BIG_DATA_CODE), obj.getStr(BaseFields.BIG_DATA_ITEMS_VALUE));
                        }
                        object.set(BaseFields.tid, item_time);
                        object.remove(BaseFields.BIG_DATA_STATUS);
                        object.set(VehicleType.VEHICLE_TYPE, this.vehicleType);
                        object.set(BaseFields.STORE_TIME, System.currentTimeMillis());
                        collector.collect(object);
                    }
                    if (BaseFields.COMMAND_REMOTE_REQUEST.equals(object.getStr(BaseFields.command))
                            ||BaseFields.COMMAND_REMOTE_CONTROL.equals(object.getStr(BaseFields.command))) {
                        List<JSONObject> objectList = map.get(item_time);
                        for (JSONObject obj : objectList) {
                            object.set(obj.getStr(BaseFields.BIG_DATA_CODE), obj.getStr(BaseFields.BIG_DATA_ITEMS_VALUE));
                        }
                        object.set(BaseFields.tid, item_time);
                        object.set(BaseFields.REQUEST_TIME, item_time);
                        object.remove(BaseFields.BIG_DATA_STATUS);
                        object.set(VehicleType.VEHICLE_TYPE, this.vehicleType);
                        object.set(BaseFields.STORE_TIME, System.currentTimeMillis());
                        collector.collect(object);
                        ctx.output(hbaseTag, object);
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
                log.error("解析云服务提供原始数据异常，data={}",object!=null?object.toString():null);
                throw e;
            }
        }
    }

    /**
     *  合并数据流
     */
    static class ConnectData extends  ProcessFunction<JSONObject, JSONObject>{
        private transient MapState<String, JSONObject> mapState;
        private transient ValueState<JSONObject> canObj;

        private transient ListState<JSONObject> listState;

        private transient ListState<JSONObject> tempState;

        private transient MapState<String,Integer> countState;
        private transient ValueState<JSONObject> requestObj;
//        private transient ValueState<JSONObject> controlObj;
        private transient ValueState<Long> onTimerTs;

        private String NO_ACTION = "1023";

        @Override
        public void onTimer(long timestamp, ProcessFunction<JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            clearObj(out);
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder(Time.seconds(300))
                    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).build();
            MapStateDescriptor<String, JSONObject> mapState = new MapStateDescriptor<>("mapState", String.class, JSONObject.class);
            mapState.enableTimeToLive(ttlConfig);
            this.mapState = getRuntimeContext().getMapState(mapState);
            this.canObj = getRuntimeContext().getState(new ValueStateDescriptor<>("canObj",JSONObject.class));
            this.requestObj = getRuntimeContext().getState(new ValueStateDescriptor<>("requestObj",JSONObject.class));
//            this.controlObj = getRuntimeContext().getState(new ValueStateDescriptor<>("controlObj",JSONObject.class));
            this.countState = getRuntimeContext().getMapState(new MapStateDescriptor<>("countState", String.class, Integer.class));
            this.listState =  getRuntimeContext().getListState(new ListStateDescriptor<JSONObject>("listState",JSONObject.class));
            this.tempState =  getRuntimeContext().getListState(new ListStateDescriptor<JSONObject>("tempState",JSONObject.class));
            this.onTimerTs = getRuntimeContext().getState(new ValueStateDescriptor<>("onTimerTs", Long.class));
        }

        @Override
        public void processElement(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
            if(BaseFields.COMMAND_STATUS.equals(object.getStr(BaseFields.command))){
                setRequestObj(object);
                if(canObj.value()!=null&&onTimerTs.value()!=null){
                    context.timerService().deleteProcessingTimeTimer(onTimerTs.value());
                    clearObj(out);
                }
                if(!mapState.isEmpty()){
                    if(mapState.contains(object.getStr(BaseFields.tid))){
                        JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                        if(obj!=null){
                            extracted(object, obj, 0, out);
                        }else{
                            cacheCanObj(object, context);

                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                        if(obj!=null){
                            extracted(object, obj, 1, out);
                        }else{
                            cacheCanObj(object, context);

                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                        if(obj!=null){
                            extracted(object, obj, -1, out);
                        }else{
                            cacheCanObj(object, context);

                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                        if(obj!=null){
                            extracted(object, obj, 2, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                        if(obj!=null){
                            extracted(object, obj, -2, out);
                        }else{
                            cacheCanObj(object, context);
                        }
                    }else{
                        cacheCanObj(object, context);
                    }
                    Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                    while (iterator.hasNext()){
                        Map.Entry<String, JSONObject> next = iterator.next();
                        if( next!=null && next.getKey().compareTo(object.getStr(BaseFields.tid))<=0){
                            iterator.remove();
                        }
                    }

                }else{
                    cacheCanObj(object, context);
                }
            }else if(BaseFields.COMMAND_TBOX.equals(object.getStr(BaseFields.command))){
                if(canObj!=null&&canObj.value()!=null){
                    JSONObject canObject = canObj.value();
                    if(canObject!=null){
                        if(object.getStr(BaseFields.tid).equals(canObject.getStr(BaseFields.tid))){
                            extracted(canObject, object, 0, out);
                            canObj.clear();
                            onTimerTs.clear();
                        }else {
                            putGpsObj(object);
                        }
                    }else{
                        putGpsObj(object);
                    }
                }else{
                    putGpsObj(object);
                }
            }else if(BaseFields.COMMAND_REMOTE_REQUEST.equals(object.getStr(BaseFields.command))){
                boolean bool = false;
                if(object.containsKey(BaseFields.DISCOUNT_POWER)
                  ||object.containsKey(BaseFields.DISCOUNT_TIME)
                  ||object.containsKey(BaseFields.EXPENSIVE_POWER)
                  ||object.containsKey(BaseFields.EXPENSIVE_TIME)){
                    if(NO_ACTION.equals(object.getStr(BaseFields.DISCOUNT_POWER))){
                        object.set(BaseFields.DISCOUNT_POWER, StrUtil.DASHED);
                    }else{
                        bool = true;
                    }
                    if(NO_ACTION.equals(object.getStr(BaseFields.DISCOUNT_TIME))){
                        object.set(BaseFields.DISCOUNT_TIME, StrUtil.DASHED);
                    }else{
                        bool = true;
                    }
                    if(NO_ACTION.equals(object.getStr(BaseFields.EXPENSIVE_POWER))){
                        object.set(BaseFields.EXPENSIVE_POWER, StrUtil.DASHED);
                    }else{
                        bool = true;
                    }
                    if(NO_ACTION.equals(object.getStr(BaseFields.EXPENSIVE_TIME))){
                        object.set(BaseFields.EXPENSIVE_TIME, StrUtil.DASHED);
                    }else{
                        bool = true;
                    }
                }
                if(bool){
                    listState.add(object);
                }
            } /*else if(BaseFields.COMMAND_REMOTE_CONTROL.equals(object.getStr(BaseFields.command))){
//                if(controlObj!=null&&null!=controlObj.value()){
//                    JSONObject obj = controlObj.value();
//                    if(obj!=null){
//                        if(canObj!=null&&canObj.value()!=null){
//                            clearObj(out);
//                        }else{
//                            clearRequestObj(obj);
//                            out.collect(obj);
//                        }
//                    }
//                }
                controlObj.update(object);
            }*/
        }

        private void cacheCanObj(JSONObject object, ProcessFunction<JSONObject, JSONObject>.Context context) throws IOException {
            canObj.update(object);
            long ts = context.timerService().currentProcessingTime() + 8000L;
            onTimerTs.update(ts);
            context.timerService().registerProcessingTimeTimer(ts);
        }

        private void extracted(JSONObject object, JSONObject obj, int value, Collector<JSONObject> out) throws IOException {
            object.set(BaseFields.LONGITUDE, obj.getStr(BaseFields.LONGITUDE));
            object.set(BaseFields.LATITUDE, obj.getStr(BaseFields.LATITUDE));
            object.set(BaseFields.MATCH_ERROR, value);
            extracted(object);
            collectCanObj(object, out);
        }

        private void collectCanObj(JSONObject object, Collector<JSONObject> out) throws IOException {
            if(requestObj!=null&&null!=requestObj.value()){
                JSONObject obj = requestObj.value();
                if(obj!=null){
                    if((object.getLong(BaseFields.tid) - obj.getLong(BaseFields.tid))>120000){
                        requestObj.clear();
                    }
                }
            }
            /*if(controlObj!=null&&null!=controlObj.value()){
                JSONObject obj = controlObj.value();
                if(obj!=null){
                    if((object.getLong(BaseFields.tid) - obj.getLong(BaseFields.tid))>120000){
                        controlObj.clear();
                    }
                }
            }*/
            out.collect(object);
        }

        private void extracted(JSONObject object) throws IOException {
            clearRequestObj(object);
//            clearControlObj(object);
        }

        private void setRequestObj(JSONObject object) throws Exception {
            if(listState!=null){
                Iterable<JSONObject> iterable = listState.get();
                if(iterable!=null){
                    Iterator<JSONObject> it = iterable.iterator();
                    if(it!=null&&it.hasNext()){
                        JSONObject request = it.next();
                        if(request!=null){
                            object.set(BaseFields.DISCOUNT_POWER, request.getStr(BaseFields.DISCOUNT_POWER));
                            object.set(BaseFields.DISCOUNT_TIME, request.getStr(BaseFields.DISCOUNT_TIME));
                            object.set(BaseFields.EXPENSIVE_POWER, request.getStr(BaseFields.EXPENSIVE_POWER));
                            object.set(BaseFields.EXPENSIVE_TIME, request.getStr(BaseFields.EXPENSIVE_TIME));
                            object.set(BaseFields.REQUEST_TIME, request.getStr(BaseFields.REQUEST_TIME));
                            if(countState!=null&&!countState.isEmpty()){
                                Integer integer = countState.get(request.getStr(BaseFields.REQUEST_TIME));
                                if(integer!=null&&integer.intValue()>0){
                                    countState.put(request.getStr(BaseFields.REQUEST_TIME),integer.intValue()+1);
                                    if(integer.intValue()>10){
                                        countState.remove(request.getStr(BaseFields.REQUEST_TIME));
                                        tempState.clear();
                                        while (it.hasNext()){
                                            tempState.add(it.next());
                                        }
                                        listState.clear();
                                        if(tempState!=null){
                                            Iterable<JSONObject> cacheIt = tempState.get();
                                            if(cacheIt!=null){
                                                Iterator<JSONObject> iterator = cacheIt.iterator();
                                                if(iterator!=null){
                                                    while (iterator.hasNext()){
                                                        listState.add(iterator.next());
                                                    }
                                                    tempState.clear();
                                                }
                                            }
                                        }
                                    }
                                }else{
                                    countState.put(request.getStr(BaseFields.REQUEST_TIME),1);
                                }
                            }else{
                                countState.put(request.getStr(BaseFields.REQUEST_TIME),1);
                            }
                        }
                    }
                }
            }
        /*if(requestObj!=null&&requestObj.value()!=null){
            JSONObject request = requestObj.value();
            if(request!=null){
                object.set(BaseFields.DISCOUNT_POWER, request.getStr(BaseFields.DISCOUNT_POWER));
                object.set(BaseFields.DISCOUNT_TIME, request.getStr(BaseFields.DISCOUNT_TIME));
                object.set(BaseFields.EXPENSIVE_POWER, request.getStr(BaseFields.EXPENSIVE_POWER));
                object.set(BaseFields.EXPENSIVE_TIME, request.getStr(BaseFields.EXPENSIVE_TIME));
                object.set(BaseFields.REQUEST_TIME, request.getStr(BaseFields.REQUEST_TIME));
                if((object.getLong(BaseFields.tid) - request.getLong(BaseFields.tid))>120000){
                    requestObj.clear();
                }
            }
        }*/
        }

        private void clearRequestObj(JSONObject object) throws IOException {
            if(requestObj!=null&&requestObj.value()!=null){
                JSONObject request = requestObj.value();
                if(request!=null){
                    object.set(BaseFields.DISCOUNT_POWER, request.getStr(BaseFields.DISCOUNT_POWER));
                    object.set(BaseFields.DISCOUNT_TIME, request.getStr(BaseFields.DISCOUNT_TIME));
                    object.set(BaseFields.EXPENSIVE_POWER, request.getStr(BaseFields.EXPENSIVE_POWER));
                    object.set(BaseFields.EXPENSIVE_TIME, request.getStr(BaseFields.EXPENSIVE_TIME));
                    object.set(BaseFields.REQUEST_TIME, request.getStr(BaseFields.REQUEST_TIME));
//                    requestObj.clear();
                }
            }
        }

       /* private void clearControlObj(JSONObject object) throws IOException {
            if(controlObj!=null&&controlObj.value()!=null){
                JSONObject control = controlObj.value();
                if(control!=null){
                    object.set("LowResp", control.getStr("LowResp"));
                    object.set("remote_5051002", control.getStr("remote_5051002"));
//                    controlObj.clear();
                }
            }
        }*/

        private void putGpsObj(JSONObject obj) throws Exception {
            JSONObject object = new JSONObject();
            object.set(BaseFields.LONGITUDE, obj.getStr(BaseFields.LONGITUDE));
            object.set(BaseFields.LATITUDE, obj.getStr(BaseFields.LATITUDE));
            object.set(BaseFields.tid, obj.getStr(BaseFields.tid));
            object.set(BaseFields.vin, obj.getStr(BaseFields.vin));
            mapState.put(object.getStr(BaseFields.tid), object);
        }

        private void clearObj(Collector<JSONObject> out) throws Exception {
            if(canObj!=null&&canObj.value()!=null){
                JSONObject object = canObj.value();
                if(object!=null){
                    if(!mapState.isEmpty()){
                        if(mapState.contains(object.getStr(BaseFields.tid))){
                            JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                            if(obj!=null) {
                                extracted(object,obj,0,out);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                            if(obj!=null) {
                                extracted(object,obj,1,out);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                            if(obj!=null) {
                                extracted(object,obj,-1,out);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                            if(obj!=null) {
                                extracted(object,obj,2,out);
                            }
                        }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                            JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                            if(obj!=null) {
                                extracted(object,obj,-2,out);
                            }
                        }
                        Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                        while (iterator.hasNext()){
                            Map.Entry<String, JSONObject> next = iterator.next();
                            if( next!=null && (next.getKey().compareTo(object.getStr(BaseFields.tid))<=0)){
                                iterator.remove();
                            }
                        }
                    }
                    extracted(object);
                    collectCanObj(object, out);
                    canObj.clear();
                    onTimerTs.clear();
                }
            }
        }
    }
}